package dubbo.server4.impl;
import java.io.IOException;
import java.net.URL;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import co.paralleluniverse.fibers.*;
import co.paralleluniverse.strands.Strand;
import co.paralleluniverse.strands.SuspendableCallable;
import co.paralleluniverse.strands.SuspendableRunnable;
import co.paralleluniverse.strands.channels.Channels;
import co.paralleluniverse.strands.channels.IntChannel;
import co.paralleluniverse.strands.concurrent.CountDownLatch;

/**
 * Increasing-Echo Quasar Example
 *
 * @author circlespainter
 */
public class QuasarIncreasingTest {
    /**
     * 测试传统线程池和协程的效率
     *
     * @return
     */
    public static String compare() {
        int count = 1000;
        try {
            testThreadpool(count);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        testFiber(count);
        return null;
    }

    private static void testFiber(int count) {
        final CountDownLatch latch = new CountDownLatch(count);
        final LongAdder latency = new LongAdder();
        long t = System.currentTimeMillis();
        for (int i = 0; i < count; i++) {
            new Fiber<Void>("Caller", new SuspendableRunnable() {
                @Override
                public void run() throws SuspendExecution, InterruptedException {
                    long start = System.currentTimeMillis();
                    m1();
                    start = System.currentTimeMillis() - start;
                    latency.add(start);
                    latch.countDown();
                }
            }).start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        t = System.currentTimeMillis() - t;
        long l = latency.longValue() / count;
        System.out.println("fiber took: " + t + ", latency: " + l + " ms");
    }

    @Suspendable
    private static void  m1() throws InterruptedException, SuspendExecution {
        String m = "m1";
        // System.out.println("m1 begin");
        m = m2();
        // System.out.println("m1 end");
        // System.out.println(m);
    }

    private static String m2() throws SuspendExecution, InterruptedException {
        String m = m3();
        Strand.sleep(1000);//模拟IO线程阻塞
        return m;
    }

    // or define in META-INF/suspendables
    @Suspendable
    private static String m3() {
        List l = Stream.of(1, 2, 3).filter(i -> i % 2 ==
                0).collect(Collectors.toList());
        return  l.toString();
    }

    private static void testThreadpool(int count) throws InterruptedException {
        final CountDownLatch latch = new CountDownLatch(count);
        ExecutorService es = Executors.newFixedThreadPool(200);
        LongAdder latency = new LongAdder();
        long t = System.currentTimeMillis();
        for (int i = 0; i < count; i++) {
            es.submit(() -> {
                long start = System.currentTimeMillis();
                try {
                    m1();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (SuspendExecution suspendExecution) {
                    suspendExecution.printStackTrace();
                }
                start = System.currentTimeMillis() - start;
                latency.add(start);
                latch.countDown();
            });
        }
        latch.await();
        t = System.currentTimeMillis() - t;
        long l = latency.longValue() / count;
        System.out.println("thread pool took: " + t + ", latency: " + l + " ms");
        es.shutdownNow();
    }
    static public Integer doAll() throws ExecutionException, InterruptedException {
        final IntChannel increasingToEcho = Channels.newIntChannel(0); // Synchronizing channel (buffer = 0)
        final IntChannel echoToIncreasing = Channels.newIntChannel(0); // Synchronizing channel (buffer = 0)

        Fiber<Integer> increasing = new Fiber<>("INCREASER", new SuspendableCallable<Integer>() { @Override public Integer run() throws SuspendExecution, InterruptedException {
            ////// The following is enough to test instrumentation of synchronizing methods
            // synchronized(new Object()) {}

            int curr = 0;
            for (int i = 0; i < 10 ; i++) {
                Fiber.sleep(10);
                System.out.println("INCREASER sending: " + curr);
                increasingToEcho.send(curr);
                curr = echoToIncreasing.receive();
                System.out.println("INCREASER received: " + curr);
                curr++;
                System.out.println("INCREASER now: " + curr);
            }
            System.out.println("INCREASER closing channel and exiting");
            increasingToEcho.close();
            return curr;
        } }).start();

        Fiber<Void> echo = new Fiber<Void>("ECHO", new SuspendableRunnable() { @Override public void run() throws SuspendExecution, InterruptedException {
            Integer curr;
            while (true) {
                Fiber.sleep(1000);
                curr = increasingToEcho.receive();
                System.out.println("ECHO received: " + curr);

                if (curr != null) {
                    System.out.println("ECHO sending: " + curr);
                    echoToIncreasing.send(curr);
                } else {
                    System.out.println("ECHO detected closed channel, closing and exiting");
                    echoToIncreasing.close();
                    return;
                }
            }
        } }).start();

        try {
            increasing.join();
            echo.join();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return increasing.get();
    }
   public static void doAll3(){
       new Fiber<Void>("Caller1", new SuspendableRunnable() {
            @Override
            public void run() throws SuspendExecution, InterruptedException {
                new Fiber<Void>("Caller2", new SuspendableRunnable() {
                    @Override
                    public void run() throws SuspendExecution, InterruptedException {
                        test1("Caller2");
                    }
                }).start();
                test1("Caller1");
            }
        }).start();
    }
    static void test1(String taskName){
        Thread cu = Thread.currentThread();
        System.out.println(taskName + ":IO前 当前线程 " + cu.getId() + " 时间 " + new Date().toLocaleString());
        try {
            new URL("http://www.baidu.com").openConnection().getContent();
        } catch (IOException e) {
        }
        System.out.println(taskName + ":IO后 当前线程 " + cu.getId() + " 时间 " + new Date().toLocaleString());
    }

    /**
     * 测试单线程下多携程情况下，任务是否遇到IO阻塞会自动让出线程资源
     * @return
     */
    public static String recycle() {
        //设置1个线程，达到演示多个任务遇到IO阻塞时Quasar调度器将线程自动切换任务轮流运行
        ExecutorService es = Executors.newFixedThreadPool(1);
        //FiberExecutorScheduler sc = new FiberExecutorScheduler("t", es);
       FiberForkJoinScheduler sc=new FiberForkJoinScheduler("f",1);
        Fiber<Void> f1 = new Fiber<Void>("Caller1",sc,  new SuspendableRunnable() {
            @Override
            public void run() throws SuspendExecution, InterruptedException {
                blockTask("Caller1");
            }
        }).start();
        Fiber<Void> f2 = new Fiber<Void>("Caller2",sc,new SuspendableRunnable() {
            @Override
            public void run() throws SuspendExecution, InterruptedException {
                blockTask("Caller2");
            }
        }).start();
        Fiber<Void> f3 = new Fiber<Void>("Caller3", sc, new SuspendableRunnable() {
            @Override
            public void run() throws SuspendExecution, InterruptedException {
                blockTask("Caller3");
            }
        }).start();
        return null;
    }
    public static void blockTask(String taskName){
        for (int i = 0; i < 5; i++) {
            Thread cu = Thread.currentThread();
            System.out.println(taskName + ":运行位置前  " + i + " 当前线程 " + cu.getId() + " 时间 " + new Date().toLocaleString());
            try {
                //Fiber.sleep(1000);
                //Strand.sleep(1000);//模拟IO线程阻塞
                new URL("http://www.baidu.com").openConnection().getContent();
				/*HttpGet request = new HttpGet("http://localhost:8771/kilim/sendAndReceive");
				// 获取当前客户端对象
				HttpClient httpClient = HttpClients.createDefault();
				// 通过请求对象获取响应对象
				HttpResponse response = httpClient.execute(request);
				// 判断网络连接状态码是否正常(0--200都数正常)
				if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
					String result = EntityUtils.toString(response.getEntity(), "utf-8");
					String rString = result;
				}*/
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println(taskName + ":运行位置后  " + i + " 当前线程 " + cu.getId() + " 时间 " + new Date().toLocaleString());
        }
    }
  /*  static public void main(String[] args) throws ExecutionException, InterruptedException {
        doAll();
    }*/
}